/*
 * delta_phase.c
 *
 *  Created on: Jul 28, 2014
 *      Author: fumin
 */
#include "destor.h"
#include "jcr.h"
#include "backup.h"
#include "storage/containerstore.h"
#include <openssl/sha.h>
#include <openssl/md5.h>
#include "storage/containerstore.h"
#include "xdelta/xdelta3.h"
#include "utils/lru_cache.h"
#include "utils/threadpool.h"

#define MAXCHUNKSIZE 4096 * 64

#define DELTASCHEME 1 // 0 dDelta, 1 xDelta, 2 zdelta
#define DELTACHECK 0  // check for restore
#define DELTASTAGE4 0

#define SLIDE1(m, fp, bufPos, buf) \
	do                             \
	{                              \
		unsigned char om;          \
		u_int64_t x;               \
		if (++bufPos >= size)      \
			bufPos = 0;            \
		om = buf[bufPos];          \
		buf[bufPos] = m;           \
		fp ^= U[om];               \
		x = fp >> shift;           \
		fp <<= 8;                  \
		fp |= m;                   \
		fp ^= T[x];                \
	} while (0)

typedef unsigned int uint32_t;
typedef unsigned char uint8_t;
typedef unsigned short uint16_t;
typedef unsigned int u_int32_t;

typedef unsigned long long int UINT64;

extern struct
{
	/* accessed in dedup phase */
	struct container *container_buffer;
	/* In order to facilitate sampling in container,
	 * we keep a queue for chunks in container buffer. */
	GSequence *chunks;
} storage_buffer;

// extern void update_score(struct chunk * c, double score);

static pthread_t delta_t;

struct chunk_nth
{
	struct chunk *c;
	int nth;
};

int ComputeDelta(struct chunk *basChunk, struct chunk *srcChunk)
{
	if (basChunk->size < 1024 || srcChunk->size < 512)
	{
		srcChunk->delta = NULL;
		return 0;
	}

	unsigned char deltaData[128 * 1024]; //[128*1024];
	uint32_t dSize;

#if DELTASCHEME == 1
	//========xDelta encoding...
	// ComJob.SimilDataSize += srcChunk->nChunkSize;
	// gettimeofday(&ComJob.deltaStart, NULL);
	int ret = xd3_encode_memory((uint8_t *)srcChunk->data, (usize_t)srcChunk->size,
								(uint8_t *)basChunk->data, (usize_t)basChunk->size,
								(uint8_t *)deltaData, (usize_t *)&dSize,
								(usize_t)srcChunk->size, (0 << XD3_COMPLEVEL_SHIFT) & XD3_COMPLEVEL_MASK);
	/*gettimeofday(&ComJob.deltaEND, NULL);
	ComJob.tEncoding += (ComJob.deltaEND.tv_sec - ComJob.deltaStart.tv_sec)*1000 +
						(ComJob.deltaEND.tv_usec - ComJob.deltaStart.tv_usec)/1000.0;
	*/
	if (ret != 0)
	{
		return 0;
	}
#if DELTACHECK
	u_int32_t rsize;
	unsigned char restore[MAXCHUNKSIZE];
	// gettimeofday(&ComJob.deltaStart, NULL);
	int r2 = xd3_decode_memory((uint8_t *)deltaData, (usize_t)dSize,
							   (uint8_t *)basChunk->data, (usize_t)basChunk->size,
							   (uint8_t *)restore, (usize_t *)&rsize, (usize_t)srcChunk->size, 1);
	/*gettimeofday(&ComJob.deltaEND, NULL);
	ComJob.tDecoding += (ComJob.deltaEND.tv_sec - ComJob.deltaStart.tv_sec)*1000 +
						(ComJob.deltaEND.tv_usec - ComJob.deltaStart.tv_usec)/1000.0;
	*/
	// if(rsize != srcChunk->size)
	//	exit(1);

	if (memcmp(srcChunk->data, restore, rsize) != 0)
	{
		printf("xDelta decode error!!!\n");
		assert(memcmp(srcChunk->data, restore, rsize) == 0);
	}
#endif
#endif

	if (dSize > srcChunk->size)
	{
		srcChunk->delta = NULL;
		return 0;
	}

	srcChunk->delta = (struct delta *)malloc(sizeof(struct delta));

	memcpy(srcChunk->delta->basefp, basChunk->fp, sizeof(fingerprint));
	srcChunk->delta->baseid = basChunk->id;
	srcChunk->delta->base_size = basChunk->size;
	srcChunk->delta->size = dSize;
	srcChunk->delta->data = (unsigned char *)malloc(dSize);
	memcpy(srcChunk->delta->data, deltaData, dSize);

	return 1;
}

int64_t aupadj, alochunk, judgenum;

void delta_left_adjchunk(struct chunk *c)
{
	int ContinueCompute = 1;
	struct segment *s = new_segment();
	g_sequence_append(s->chunks, c);
	while (1)
	{
		c = sync_queue_get_top(matching_queue);
		assert(!CHECK_CHUNK(c, CHUNK_FILE_START) &&
			   !CHECK_CHUNK(c, CHUNK_FILE_END) &&
			   !CHECK_CHUNK(c, CHUNK_SEGMENT_START) &&
			   !CHECK_CHUNK(c, CHUNK_SEGMENT_END));
		if (CHECK_CHUNK(c, CHUNK_DUPLICATE) ||
			c->baseChunk && !CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_LEFT))
			break;
		c = sync_queue_pop(matching_queue);
		g_sequence_append(s->chunks, c);
	}

	int len = g_sequence_get_length(s->chunks);
	GSequenceIter *begin = g_sequence_get_begin_iter(s->chunks);
	assert(g_sequence_iter_get_position(begin) == 0);
	GSequenceIter *back = g_sequence_get_iter_at_pos(s->chunks, len - 1);

	TIMER_DECLARE(1);
	TIMER_BEGIN(1);
	while (1)
	{
		struct chunk *ck = g_sequence_get(back);

		if (ck->baseChunk)
		{
			if (ContinueCompute && ComputeDelta(ck->baseChunk, ck))
			{
				int compressed_size = ck->size - ck->delta->size;
				double score = compressed_size * 1.0 / ck->size;

				if (score > 0.9)
				{
					aupadj++;
					jcr.total_compressed_size += compressed_size;
					jcr.total_size_for_deltacompression += ck->size;
					jcr.number_of_chunks_has_delta++;
				}
				else
				{
					ck->delta = NULL;
					free_chunk(ck->baseChunk);
					ck->baseChunk = NULL;
					ck->type = CHUNK_TYPE_UNIQUE;
					ContinueCompute = 0;
				}
			}
			else
			{
				ck->delta = NULL;
				free_chunk(ck->baseChunk);
				ck->baseChunk = NULL;
				ck->type = CHUNK_TYPE_UNIQUE;
				ContinueCompute = 0;
				alochunk++;
			}
		}
		if (back == begin)
			break;
		back = g_sequence_iter_prev(back);
	}
	TIMER_END(1, jcr.delta_time);

	GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
	begin = g_sequence_get_begin_iter(s->chunks);
	while (begin != end)
	{
		struct chunk *c = g_sequence_get(begin);
		sync_queue_push(delta_queue, c);
		g_sequence_remove(begin);
		begin = g_sequence_get_begin_iter(s->chunks);
	}

	free_segment(s);
	s = NULL;
}

void *delta_thread(void *arg)
{
	int isRightChunk = 0;
	while (1)
	{
		struct chunk *c = sync_queue_pop(matching_queue);

		if (c == NULL)
			/* backup job finish */
			break;

		// sync_queue_push(delta_queue, c);
		// continue;

		if (CHECK_CHUNK(c, CHUNK_FILE_START) ||
			CHECK_CHUNK(c, CHUNK_FILE_END) ||
			CHECK_CHUNK(c, CHUNK_SEGMENT_START) ||
			CHECK_CHUNK(c, CHUNK_SEGMENT_END))
		{
			isRightChunk = 0;
			sync_queue_push(delta_queue, c);
			continue;
		}

		if (CHECK_CHUNK(c, CHUNK_DUPLICATE) || c->size <= destor.chunk_min_size)
		{
			isRightChunk = 0;
			sync_queue_push(delta_queue, c);
			continue;
		}

		if (c->baseChunk)
		{
			if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_LEFT))
			{
				delta_left_adjchunk(c);
				isRightChunk = 0;
				continue;
			}

			if (isRightChunk == 1 && CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_RIGHT))
			{
				c->delta = NULL;
				free_chunk(c->baseChunk);
				c->baseChunk = NULL;
				c->type = CHUNK_TYPE_UNIQUE;
				alochunk++;
				sync_queue_push(delta_queue, c);
				continue;
			}

			TIMER_DECLARE(1);
			TIMER_BEGIN(1);
			if (ComputeDelta(c->baseChunk, c))
			{
				int compressed_size = c->size - c->delta->size;
				double score = compressed_size * 1.0 / c->size;
				int cancel = 0;

				if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE))
				{
					if (score > 0.9)
					{
						assert(compressed_size <= c->size);
						aupadj++;
						jcr.total_compressed_size += compressed_size;
						jcr.total_size_for_deltacompression += c->size;
						jcr.number_of_chunks_has_delta++;
					}
					else
					{
						c->delta = NULL;
						free_chunk(c->baseChunk);
						c->baseChunk = NULL;
						cancel = 1;
						c->type = CHUNK_TYPE_UNIQUE;
					}
				}
				else
				{
					jcr.total_compressed_size += compressed_size;
					jcr.total_size_for_deltacompression += c->size;
					jcr.number_of_chunks_has_delta++;
				}
			}
			else
			{
				c->delta = NULL;
				free_chunk(c->baseChunk);
				c->baseChunk = NULL;
				jcr.unproper_delta_comp_num++;
				c->type = CHUNK_TYPE_UNIQUE;
			}
			if (c->delta == NULL && CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_RIGHT))
				isRightChunk = 1;
			else
				isRightChunk = 0;
			destor.now_delta_num--;
			TIMER_END(1, jcr.delta_time);
		}
		sync_queue_push(delta_queue, c);
	}
	sync_queue_term(delta_queue);

	return NULL;
}

int get_delta_segment(struct segment *s, GSequenceIter **BaseArray, GSequenceIter **BackupArray, int BaseLen)
{
	int i;
	int NewLen = 0;
	int len = g_sequence_get_length(s->chunks);
	for (i = 0; i < BaseLen; ++i)
	{
		struct chunk *c = g_sequence_get(BaseArray[i]);
		struct chunk *ck = NULL;
		int index = g_sequence_iter_get_position(BaseArray[i]);
		GSequenceIter *cur = BaseArray[i];

		if (c->delta)
		{
			if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_LEFT))
			{
				cur = g_sequence_iter_prev(cur);
				index--;
				int off = 0;
				while (index >= 0 && off++ < destor.adjchunk_match_length)
				{
					ck = g_sequence_get(cur);
					if ((ck->level == 0 && !CHECK_CHUNK(ck, CHUNK_UNIQUE)) || CHECK_CHUNK(ck, CHUNK_ADJ_UNIQUE_RIGHT))
					{
						break;
					}
					if (CHECK_CHUNK(ck, CHUNK_ADJ_UNIQUE_LEFT))
					{
						assert(ck->baseChunk);
						assert(ck->level != 0);
						BackupArray[NewLen++] = cur;
						break;
					}
					cur = g_sequence_iter_prev(cur);
					index--;
				}
			}
			else if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_RIGHT))
			{
				cur = g_sequence_iter_next(cur);
				index++;
				int off = 0;
				while (index < len && off++ < destor.adjchunk_match_length)
				{
					ck = g_sequence_get(cur);
					if ((ck->level == 0 && !CHECK_CHUNK(ck, CHUNK_UNIQUE)) || CHECK_CHUNK(ck, CHUNK_ADJ_UNIQUE_LEFT))
					{
						break;
					}
					if (CHECK_CHUNK(ck, CHUNK_ADJ_UNIQUE_RIGHT))
					{
						assert(ck->baseChunk);
						assert(ck->level != 0);
						BackupArray[NewLen++] = cur;
						break;
					}
					cur = g_sequence_iter_next(cur);
					index++;
				}
			}
		}
	}

	for (i = 0; i < NewLen; ++i)
	{
		BaseArray[i] = BackupArray[i];
	}

	return NewLen;
}

void *new_compute_thread(void *arg)
{
	struct chunk *c = (struct chunk *)arg;
	assert(c->baseChunk);
	assert(c->delta == NULL);
	assert(c->level != 0);
	c->level = 0;
	if (ComputeDelta(c->baseChunk, c))
	{
		int compressed_size = c->size - c->delta->size;
		double score = compressed_size * 1.0 / c->size;
		if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE))
		{
			if (score > 0.9)
			{
				aupadj++;
				jcr.total_compressed_size += compressed_size;
				jcr.total_size_for_deltacompression += c->size;
				jcr.number_of_chunks_has_delta++;
			}
			else
			{
				c->delta = NULL;
				free_chunk(c->baseChunk);
				c->baseChunk = NULL;
				c->type = CHUNK_TYPE_UNIQUE;
				judgenum++;
			}
		}
		else
		{
			jcr.total_compressed_size += compressed_size;
			jcr.total_size_for_deltacompression += c->size;
			jcr.number_of_chunks_has_delta++;
		}
	}
	else
	{
		c->delta = NULL;
		free_chunk(c->baseChunk);
		c->baseChunk = NULL;
		jcr.unproper_delta_comp_num++;
		c->type = CHUNK_TYPE_UNIQUE;
		if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE))
			judgenum++;
	}
}

void *delta_thread_adjacent(void *arg)
{
	int thread_num = destor.thread_num;
	threadpool_t *thp = threadpool_create(thread_num, 2000);
	while (1)
	{
		struct chunk *c = sync_queue_pop(matching_queue);

		if (c == NULL)
		{
			break;
		}

		// sync_queue_push(delta_queue, c);
		// continue;

		struct segment *s = new_segment();
		assert(CHECK_CHUNK(c, CHUNK_SEGMENT_START));
		free_chunk(c);
		int delta_num = 0;
		c = sync_queue_pop(matching_queue);
		while (!(CHECK_CHUNK(c, CHUNK_SEGMENT_END)))
		{
			g_sequence_append(s->chunks, c);
			if (!CHECK_CHUNK(c, CHUNK_FILE_START) && !CHECK_CHUNK(c, CHUNK_FILE_END))
				s->chunk_num++;
			if (c->baseChunk)
				delta_num++;
			if (CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_LEFT) && CHECK_CHUNK(c, CHUNK_ADJ_UNIQUE_RIGHT))
				assert(0);
			c = sync_queue_pop(matching_queue);
		}

		free_chunk(c);
		if (delta_num != 0)
		{
			int len = g_sequence_get_length(s->chunks), i;
			GSequenceIter **BaseArray = (GSequenceIter **)malloc(sizeof(GSequenceIter *) * len);
			GSequenceIter **BackupArray = (GSequenceIter **)malloc(sizeof(GSequenceIter *) * len);
			for (i = 0; i < len; ++i)
				BaseArray[i] = BackupArray[i] = NULL;

			int BaseLen = 0;

			GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
			GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
			for (; iter != end; iter = g_sequence_iter_next(iter))
			{
				struct chunk *c = g_sequence_get(iter);
				if (c->level == 1)
				{
					assert(c->baseChunk);
					BaseArray[BaseLen++] = iter;
				}
			}

			TIMER_DECLARE(1);
			TIMER_BEGIN(1);
			while (BaseLen)
			{
				for (i = 0; i < BaseLen; ++i)
				{
					struct chunk *c = g_sequence_get(BaseArray[i]);
					threadpool_add(thp, new_compute_thread, (void *)c);
				}
				wait_threadpool(thp, BaseLen);
				BaseLen = get_delta_segment(s, BaseArray, BackupArray, BaseLen);
			}

			TIMER_END(1, jcr.delta_time);

			iter = g_sequence_get_begin_iter(s->chunks);
			end = g_sequence_get_end_iter(s->chunks);
			for (; iter != end; iter = g_sequence_iter_next(iter))
			{
				struct chunk *c = g_sequence_get(iter);
				if (c->level > 0)
				{
					assert(c->baseChunk);
					assert(c->delta == NULL);
					c->level = 0;
					c->delta = NULL;
					free_chunk(c->baseChunk);
					c->baseChunk = NULL;
					c->type = CHUNK_TYPE_UNIQUE;
					alochunk++;
				}
			}

			free(BaseArray);
			free(BackupArray);
		}
		send_segment_filter(s);
		free_segment(s);
		s = NULL;
	}
	threadpool_destroy(thp);
	sync_queue_term(delta_queue);
	return NULL;
}

void compute_thread(void *arg)
{
	struct chunk *c = (struct chunk *)arg;
	assert(c->baseChunk);
	c->level = 0;
	if (ComputeDelta(c->baseChunk, c))
	{
		jcr.number_of_chunks_has_delta++;
		int compressed_size = c->size - c->delta->size;
		jcr.total_compressed_size += compressed_size;
		jcr.total_size_for_deltacompression += c->size;
	}
	else
	{
		c->delta = NULL;
		free_chunk(c->baseChunk);
		c->baseChunk = NULL;
		jcr.unproper_delta_comp_num++;
		c->type = CHUNK_TYPE_UNIQUE;
	}

	return NULL;
}

void send_segment_filter(struct segment *s)
{
	struct chunk *ss = new_chunk(0);
	SET_CHUNK(ss, CHUNK_SEGMENT_START);
	sync_queue_push(delta_queue, ss);

	GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
	GSequenceIter *begin = g_sequence_get_begin_iter(s->chunks);
	while (begin != end)
	{
		struct chunk *c = g_sequence_get(begin);
		sync_queue_push(delta_queue, c);
		g_sequence_remove(begin);
		begin = g_sequence_get_begin_iter(s->chunks);
	}

	struct chunk *se = new_chunk(0);
	SET_CHUNK(se, CHUNK_SEGMENT_END);
	sync_queue_push(delta_queue, se);
	s->chunk_num = 0;
}

void *delta_thread_pool(void *arg)
{
	int thread_num = destor.thread_num;

	threadpool_t *thp = threadpool_create(thread_num, 2000);

	while (1)
	{
		struct chunk *c = sync_queue_pop(matching_queue);

		if (c == NULL)
		{
			break;
		}

		struct segment *s = new_segment();
		int delta_num = 0;
		assert(CHECK_CHUNK(c, CHUNK_SEGMENT_START));
		free_chunk(c);

		c = sync_queue_pop(matching_queue);
		while (!(CHECK_CHUNK(c, CHUNK_SEGMENT_END)))
		{
			g_sequence_append(s->chunks, c);
			if (!CHECK_CHUNK(c, CHUNK_FILE_START) && !CHECK_CHUNK(c, CHUNK_FILE_END))
				s->chunk_num++;
			c = sync_queue_pop(matching_queue);
		}

		free_chunk(c);

		int task_num = 0;
		TIMER_DECLARE(1);
		TIMER_BEGIN(1);

		GSequenceIter *iter = g_sequence_get_begin_iter(s->chunks);
		GSequenceIter *end = g_sequence_get_end_iter(s->chunks);
		for (; iter != end; iter = g_sequence_iter_next(iter))
		{
			struct chunk *c = g_sequence_get(iter);
			if (c->baseChunk)
			{
				task_num++;
				threadpool_add(thp, compute_thread, (void *)c);
			}
		}
		wait_threadpool(thp, task_num);
		TIMER_END(1, jcr.delta_time);

		send_segment_filter(s);
		free_segment(s);
		s = NULL;
	}
	threadpool_destroy(thp);
	sync_queue_term(delta_queue);
	return NULL;
}

void start_delta_phase()
{
	delta_queue = sync_queue_new(1000);
	if (destor.enable_match_adjacent)
	{
		if (destor.thread_num == 1)
			pthread_create(&delta_t, NULL, delta_thread, NULL);
		else
			pthread_create(&delta_t, NULL, delta_thread_adjacent, NULL);
	}
	else if (destor.thread_pool)
		pthread_create(&delta_t, NULL, delta_thread_pool, NULL);
	else
		pthread_create(&delta_t, NULL, delta_thread, NULL);
}

void stop_delta_phase()
{
	pthread_join(delta_t, NULL);
	printf("The CHUNK_DUPADJ_UNIQUE %d\n", aupadj);
	printf("The alone chunk %d\n", alochunk);
	printf("The judge %d\n", judgenum);
}
